package com.csthink.mr.wordcount;

import com.csthink.utils.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

import java.io.IOException;

/**
 * hadoop jar hadoop-study-1.0-SNAPSHOT.jar com.csthink.mr.wordcount.WordCountDriver <input path> <output path>
 * @author <a href="mailto:csthink@icloud.com">Mars</a>
 * @since 2024-04-12 17:02
 */
public class WordCountDriver {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //String input = "data/wc.txt";
        //String output = "out";

        Configuration conf = new Configuration();

        String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
        if (otherArgs.length < 2) {
            System.err.println("Usage: WordCountDriver <in> [<in>...] <out>");
            System.exit(2);
        }

        //if (args.length != 2) {
        //    System.err.println("Usage: WordCountDriver <input path> <output path>");
        //    System.exit(-1);
        //}
        String input = args[0];
        String output = otherArgs[otherArgs.length - 1];


        Job job = Job.getInstance(conf);
        job.setJobName("WordCount");

        // reduce 函数输出文件的写入目录，在运行作业前该目录是不应该存在的，否则 Hadoop 会报错并拒绝允许作业
        // 这种预防措施的目的是防止数据丢失(长时间运行的作业如果结果被意外覆盖，肯定是非常恼人的)
        FileUtils.deleteIfExists(conf, output);

        // 设置 Job 运行的主类，注意这里必须设置，否则集群中运行会报错，找不到主类
        job.setJarByClass(WordCountDriver.class);

        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        /*
         如果 mapper 产生出和 reducer 相同的类型时，不需要单独设置，如果不同，必须设置 map 函数的输出类型
         job.setMapOutputKeyClass(Text.class);
         job.setMapOutputValueClass(IntWritable.class);
        */
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 设置输入路径(可以是文件或者目录)
        //TextInputFormat.setInputPaths(job, new Path(input));
        //FileInputFormat.setInputPaths(job, new Path(input));
        // 可以多次调用 addInputPath() 来实现多路径的输入
        for (int i = 0; i < otherArgs.length - 1; ++i) {
            FileInputFormat.addInputPath(job, new Path(otherArgs[i]));
        }
        //FileInputFormat.addInputPath(job, new Path(input));

        // 指定 reduce 函数输出文件的写入目录
        FileOutputFormat.setOutputPath(job, new Path(output));

        // 提交作业并等待执行完成，参数 true 指示是否生成详细的输出，会把进度信息写到控制台
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
